package my.thesis.analytics.metrics;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Collection;
import java.util.Map;

import org.joda.time.DateTime;

import backtype.storm.metric.api.IMetricsConsumer;
import backtype.storm.metric.api.IMetricsConsumer.DataPoint;
import backtype.storm.metric.api.IMetricsConsumer.TaskInfo;
import backtype.storm.task.IErrorReporter;
import backtype.storm.task.TopologyContext;

public class MyMetricsConsumer implements IMetricsConsumer 
{
	static private String padding = "                       ";  
	
	File f;
	BufferedWriter bw;
	String outputLine = "";
	
	public void prepare(Map stormConf, Object registrationArgument,TopologyContext context, IErrorReporter errorReporter) 
	{
		DateTime now = DateTime.now();
		String timestamp = "ts-"+ now.getHourOfDay() + "_" + now.getMinuteOfHour() + "_00";
			
		//2 pcs
		f=new File("C:/Users/Public/MyMetrics_TOPO2_FULL_UNREL_" + timestamp +".txt");
		
		//f=new File("C:/Users/vaggelis/Documents/myMetrics_" + timestamp +".txt");
						
		if(!f.exists())
		{
			try
			{
				f.createNewFile();
			} catch (IOException e)
			{
				System.err.println("Error while creating file "+f.getAbsolutePath()+e.getMessage());
			}
		}
	}

	public void handleDataPoints(TaskInfo taskInfo,Collection<DataPoint> dataPoints) 
	{
		
		 StringBuilder sb = new StringBuilder();
		 	//host/task info
	        String header = String.format("%d\t%15s:%-4d\t%3d:%-11s\t",
	            taskInfo.timestamp,
	            taskInfo.srcWorkerHost, taskInfo.srcWorkerPort,
	            taskInfo.srcTaskId,
	            taskInfo.srcComponentId);
	        sb.append(header);
	        for (DataPoint p : dataPoints) 
	        {	
	        	//EDW XREIAZETI NA PROSDIORISEIS AKRIVWS ME TO ONOMA POIES METRIKES THELEIS NA GRAPSEIS GIATI ALLIWS VGAZEI DATA GIA OLA TA BOLTS 
	        	//if(p.name.equals("topHashtagCount") || p.name.equals("sameHashtagCount") || p.name.equals("testCount")|| 
	        			//p.name.equals("sentimentCount") || p.name.equals("mentionCount")|| p.name.equals("topHashtagPersist")|| p.name.equals("sameHashtagPersist") ||
	        			//p.name.equals("sentimentPersist")|| p.name.equals("mentionPersist") ||  p.name.equals("filterLanguageCount"))
	        			if(p.name.equals("sentimentCount") || p.name.equals("fieldTopo2"))
	        				
	        			
	        			{
	        	
	        	
		            sb.delete(header.length(), sb.length());
		            //metric name
		            sb.append(p.name)
		                .append(padding).delete(header.length()+23,sb.length()).append("\t")
		                //metic value
		                .append(p.value);
		            
		            System.out.println(sb.toString());
		            
		    		try 
		    		{
		                bw = new BufferedWriter(new FileWriter(f, true));
		                bw.write(sb.toString());
		                bw.newLine();
		                bw.close();
		    		} 
		    		catch (Exception e) 
		    		{
		    			System.err.println("Error writing in File");
		    		}
	        	}
	        	//else
	        	//{
	        		//do nothing
	        	//}
	        }
	}

	public void cleanup() {
		// TODO Auto-generated method stub
		
	}

}
